package cn.reach.rpc;

import cn.reach.rpc.future.DefaultFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @Author: ReachYin
 * @Date: 2025/7/2 13:45
 * @Version: v1.0.0
 * @Description: 同步获取通讯结果模板服务
 **/
public abstract class AbstractInvokeRpcService<R, P> implements IInvokeRpc<R, P> {
    private final Map<Long, DefaultFuture<R>> futureMap = new ConcurrentHashMap<Long, DefaultFuture<R>>();
    protected final Logger logger = LoggerFactory.getLogger(getClass());
    protected RpcConfig config = RpcConfig.createDefault();

    public void setConfig(RpcConfig config) {
        this.config = config;
    }

    public RpcConfig getConfig() {
        return config;
    }

    public R getResult(P param) {
        return this.getResult(config.getDefaultTimeout(), param);
    }

    public R getResult(int timeOut, P param) throws RuntimeException {
        int retryTimes = config.getRetryTimes();
        RuntimeException lastException = null;

        for (int i = 0; i <= retryTimes; i++) {
            try {
                return doGetResult(timeOut, param);
            } catch (RuntimeException e) {
                lastException = e;
                if (i < retryTimes) {
                    logger.warn("RPC调用失败，准备第{}次重试: {}", (i + 1), e.getMessage());
                    try {
                        Thread.sleep(config.getRetryInterval());
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                        throw new RuntimeException("重试被中断", ie);
                    }
                }
            }
        }

        throw lastException;
    }

    private R doGetResult(int timeOut, P param) throws RuntimeException {
        // 获取future对象
        long futureId = creatFutureId();
        DefaultFuture<R> future = new DefaultFuture<R>();
        futureMap.put(futureId, future);

        // 发送请求
        try {
            sendRequest(futureId, param);
        } catch (Exception e) {
            futureMap.remove(futureId);
            future.doReceived(null);
            logger.error("发送请求失败", e);
            throw new RuntimeException("请求出错: " + e.getMessage(), e);
        }

        // 同步等待结果
        R res;
        try {
            res = future.get(timeOut);
            if (res == null && config.isFailFast()) {
                throw new RuntimeException("请求返回结果为null");
            }
            return res;
        } catch (Exception e) {
            logger.error("请求超时或被中断, futureId={}, timeout={}", futureId, timeOut, e);
            throw new RuntimeException("请求超时或被中断: " + e.getMessage(), e);
        } finally {
            // 确保在任何情况下都清理资源
            futureMap.remove(futureId);
        }
    }


    /**
     * TCP响应时调用，通知等待的线程，返回结果
     *
     * @param futureId 之前创建的futureId
     * @param response 返回结果
     */
    public void received(long futureId, R response) {
        DefaultFuture<R> future = futureMap.get(futureId);
        if (future != null) {
            future.doReceived(response);
            futureMap.remove(futureId);
        }
    }


    /**
     * 生成唯一Id
     *
     * @return 唯一Id
     */
    public abstract long creatFutureId();

    /**
     * 发送TCP请求
     *
     * @param futureId 唯一Id
     * @param param    请求参数
     */
    public abstract void sendRequest(long futureId, P param);


}
